package com.pwc.deadletter;

import com.pwc.utils.RabbitUtil;
import com.rabbitmq.client.*;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author fantuancyh
 * @data 2021/8/21
 */
public class Consumer {
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    private static final String DEAD_EXCHANGE = "dead_exchange";

    private static final String NORMAL_QUEUE = "normal-queue";
    private static final String DEAD_QUEUE = "dead-queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitUtil.makeChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        Map<String, Object> arguments = new HashMap<>();
        //arguments.put("x-message-ttl",10*1000);
        // 队列超出限制长度
        // arguments.put("x-max-length",6);
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        arguments.put("x-dead-letter-routing-key", "lisi");

        channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments);
        channel.queueDeclare(DEAD_QUEUE, true, false, false, null);

        //绑定普通交换机和队列
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println(new String(message.getBody(), "utf-8"));
            //消息被拒绝
            if (StringUtils.equals("信息5", new String(message.getBody(), "utf-8"))) {
                channel.basicNack(message.getEnvelope().getDeliveryTag(), false, false);
            }
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("error");
        };
        System.out.println("正常queue等待接受消息");
        boolean autoAck = false;
        channel.basicConsume(NORMAL_QUEUE, autoAck, deliverCallback, cancelCallback);
        channel.basicConsume(DEAD_QUEUE, autoAck, deliverCallback, cancelCallback);
    }
}
